import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import pyspark.sql.functions as sf
 
## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
 
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
 
store_DyF = glueContext.create_dynamic_frame.from_catalog(database="awsdatalakesession", table_name="store_reference")
tran_DyF = glueContext.create_dynamic_frame.from_catalog(database="awsdatalakesession", table_name="transaction")
tran_DF=tran_DyF.toDF()
tran_DF = tran_DF.selectExpr("store_id as store_id_tran", "product_name  as product_name", "total  as total",  "date_tran  as date_tran")
store_DF=store_DyF.toDF()
join_DF = tran_DF.join(store_DF, tran_DF.store_id_tran == store_DF.store_id,"inner")
join_DF = join_DF.selectExpr("store_id as store_id", "product_name  as product_name", "total  as total",  "date_tran  as date_tran", "city as city","state as state", "territory as territory")
agg_DF=join_DF.groupBy("city").agg(sf.sum('total').alias('city_total'))
agg_DF.coalesce(2).write.parquet('s3://<your bucket name>/data/total_by_city') 
 

 
job.commit()